package com.hliushi.mq.base.comsumer;


import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * 消息的接收者
 *
 * @author llhuang10
 * @date 2021/8/11 17:11
 */
public class Consumer {

    public static void main(String[] args) {

        /**
         * DefaultMQPushConsumer   push 拉取型数据
         */
        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("group1");
        pushConsumer.setNamesrvAddr("cvm-node01:9876;cvm-node02:9876");

        // 设定消费模式: 负载均衡|广播模式
        // 默认是负载均衡模式
        pushConsumer.setMessageModel(MessageModel.BROADCASTING);

        try {
            // 订阅主题Topic和Tag,以消费订阅
            pushConsumer.subscribe("base", "Tag1 || Tag2");
            // 设置回调函数, 处理消息
            pushConsumer.registerMessageListener(new MessageListenerConcurrently() {

                /**
                 * 接收消息内容
                 * @param msgs
                 * @param context
                 * @return
                 */
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    // System.out.println("msgs = " + msgs);
                    msgs.forEach(item -> {
                        String string = new String(item.getBody());
                        System.out.println("body = " + string);
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            // 启动消费者 consumer
            pushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
